package p.ithorns.framework.logger;

import ch.qos.logback.classic.Logger;
import ch.qos.logback.classic.LoggerContext;
import ch.qos.logback.classic.spi.ILoggingEvent;
import ch.qos.logback.core.OutputStreamAppender;
import ch.qos.logback.core.spi.DeferredProcessingAware;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.slf4j.LoggerFactory;
import p.ithorns.framework.logger.appender.PipedStreamAppender;
import p.ithorns.framework.logger.stream.PipelineStream;

import java.io.*;
import java.nio.charset.StandardCharsets;
import java.util.Collection;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;

/**
 * PipedStreamLogger
 *
 * @author ithorns
 * @version 1.0.0
 * @date 2023/3/10 14:06
 */
public class PipedStreamLogger extends OutputStreamAppender<ILoggingEvent> {

    private final static org.slf4j.Logger log = LoggerFactory.getLogger(PipedStreamLogger.class);

    private final static ThreadFactory THREAD_FACTORY = new ThreadFactoryBuilder()
            .setDaemon(false)
            .setNameFormat("log-pipe-%d")
            .build();

    private final static PipedStreamAppender APPENDER;

    static {
        LoggerContext loggerContext = (LoggerContext) LoggerFactory.getILoggerFactory();
        Logger rootLogger = loggerContext.getLogger("root");
        APPENDER = (PipedStreamAppender) rootLogger.getAppender("PIPE");
    }

    private final ExecutorService daemonPool;

    private final ExecutorService consumerPool;

    private final PipelineStream stream;


    private final Map<String, Consumer<String>> consumers;

    public PipedStreamLogger() {
        this.consumers = new ConcurrentHashMap<>();
        this.daemonPool = ThreadPoolFactory.createDaemonPool("log-pipe-daemon");
        this.consumerPool = ThreadPoolFactory.create(2, 4, "log-pipe-consumer");
        this.stream = new PipelineStream();
    }

    @Override
    public void start() {
        setOutputStream(stream.getOut());
        super.start();
        this.startRead();
    }

    /**
     * 启动读
     */
    public void startRead() {
        daemonPool.execute(this::read);
    }

    /**
     * 连接消费者
     *
     * 开启一个新的线程处理日志数据回调，
     * 当管道有日志内容时，交由消费者具体实现来处理
     *
     * @param consumer -消费者
     */
    public void apply(String consumerId, final Consumer<String> consumer) {
        consumers.putIfAbsent(consumerId, consumer);
    }

    /**
     * 删除消费客户端
     *
     * @param consumer -消费者
     */
    public void remove(String consumer) {
        consumers.remove(consumer);
    }

    /**
     * 读消息
     */
    private void read() {
        String line;
        try (
                InputStreamReader readerIn = new InputStreamReader(stream.getIn());
                BufferedReader br = new BufferedReader(readerIn)
        ) {
            while ((line = br.readLine()) != null) {
                sendToConsumers(line);
            }
        } catch (Exception e) {
            e.printStackTrace();
            addError("PipedStreamLogger => read error", e);
        }
        log.info("这个读线程他结束了！！！");

        stream.createPipe();
        start();
        log.info("重新启动一次......");
    }

    private void read2() {
        try {
            InputStream in = stream.getIn();
            while (!Thread.currentThread().isInterrupted()) {
                int available = in.available();
                if (available > 0) {
                    byte[] buffer = new byte[available];
                    in.read(buffer, 0, available);
                    String line = new String(buffer, StandardCharsets.UTF_8);
                    sendToConsumers(line);
                } else {
                    TimeUnit.MILLISECONDS.sleep(2);
                }
            }
        } catch (IOException | InterruptedException e) {
            e.printStackTrace();
        }
    }

    private void sendToConsumers(final String line) {
        Collection<Consumer<String>> values = consumers.values();
        for (Consumer<String> consumer : values) {
            consumerPool.execute(() -> consumer.accept(line));
        }
    }

    @Override
    public void append(ILoggingEvent event) {
        if (!isStarted()) {
            return;
        }

        try {
            // this step avoids LBCLASSIC-139
            if (event instanceof DeferredProcessingAware) {
                ((DeferredProcessingAware) event).prepareForDeferredProcessing();
            }

            byte[] bytes = encoder.encode(event);
            OutputStream outputStream = getOutputStream();
            outputStream.write(bytes);
            outputStream.flush();
        } catch (Exception e) {
            addError("Error writing to output stream.", e);
        }
    }

    @Override
    public void stop() {
        if (null != this.stream) {
            this.stream.close();
        }
        super.stop();

        this.consumerPool.shutdown();
        this.daemonPool.shutdown();
    }

}